package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TrafficPageViewBean1;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficChIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端

        // TODO 3 读取两条流的数据数据
        // 3.1 读取页面主题数据
        String page_topic = "dwd_traffic_page_log";
        String groupId =
                "dws_traffic_ch_is_new_page_view_window";
        DataStreamSource<String> pageStream =
                env.addSource(KafkaUtil.getFlinkKafkaConsumer(page_topic,
                        groupId));

        //3.2 读取独立访客页面主题数据
        String uvTopic =
                "dwd_traffic_unique_visitor_detail";
        DataStreamSource<String> uvStream =
                env.addSource(KafkaUtil.getFlinkKafkaConsumer(uvTopic,
                        groupId));

        // TODO 4 转换两条流，存储对应的度量值
        SingleOutputStreamOperator<TrafficPageViewBean1> pageBeanStream =
                pageStream.map(new MapFunction<String, TrafficPageViewBean1>() {
                    @Override
                    public TrafficPageViewBean1 map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject page = jsonObject.getJSONObject("page");
                        String lastPageId = page.getString("last_page_id");
                        return new TrafficPageViewBean1("", "",common.getString("ch"),common
                                .getString("is_new"),0L, 1L, jsonObject.getLong("ts"));
                    }
                });

        SingleOutputStreamOperator<TrafficPageViewBean1> uvBeanStream = uvStream.map(new MapFunction<String, TrafficPageViewBean1>() {
            @Override
            public TrafficPageViewBean1 map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                return new TrafficPageViewBean1("", "",common.getString("ch"),common.getString("is_new"), 1L, 0L, jsonObject.getLong("ts"));
            }
        });

        // TODO 5 合并两条流的数据
        DataStream<TrafficPageViewBean1> unionStream =
                pageBeanStream.union(uvBeanStream);

        // TODO 6 开窗聚合
        KeyedStream<TrafficPageViewBean1, String> keyedStream = unionStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean1>forBoundedOutOfOrderness(Duration.ofSeconds(5L)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean1>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean1 element,
                                         long recordTimestamp) {
                return element.getTs();
            }
        })).keyBy(new KeySelector<TrafficPageViewBean1, String>() {
            @Override
            public String getKey(TrafficPageViewBean1 value) throws Exception {
                return value.getIsNew();
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean1> reduceStream =
                keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TrafficPageViewBean1>() {
                    @Override
                    public TrafficPageViewBean1 reduce(TrafficPageViewBean1 value1, TrafficPageViewBean1 value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        return value1;
                    }
                }, new WindowFunction<TrafficPageViewBean1,
                        TrafficPageViewBean1, String,
                        TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window,
                                      Iterable<TrafficPageViewBean1> input,
                                      Collector<TrafficPageViewBean1> out) throws Exception {
                        TrafficPageViewBean1 next = input.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        reduceStream.print("reduce>>>>>");

        // TODO 7 写出到clickhouse
        reduceStream.addSink(ClickHouseUtil.getClickHouseSink(
                "insert into dws_traffic_ch_is_new_page_view_window values(?,?,?,?,?,?,?)"));

        // TODO 8 执行任务
        env.execute(groupId);
    }
}
